// Copyright (c) 2011 SameGoal LLC.
// All Rights Reserved.
// Author: Andy Hochhaus <ahochhaus@samegoal.com>

// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

#include "net/http/server/comet/session.h"

#include <fcntl.h>
#include <gflags/gflags.h>
#include <glog/logging.h>

#include <map>
#include <string>
#include <vector>

#include "base/thread/mutex.h"
#include "net/http/server/comet/json.h"
#include "net/http/server/comet/session_manager.h"
#include "net/http/server/comet/request_helper.h"
#include "net/http/server/comet/comet.h"
#include "net/http/server/event/connection.h"

namespace sg {
namespace net {
namespace http {
namespace server {
namespace comet {

const int64 kSessionInactivityThreshold = 60 * 60;  // 1 hour
const int64 kBackChannelThreshold = 60 * 3.8;  // 3.8 mins
const int64 kNoopInterval = 30;

Session::Session(CometServer *server)
    : request_(NULL),
      timer_event_(kNoopInterval, std::bind(&Session::NoopCallback, this)),
      last_back_channel_write_activity_(), back_channel_start_(),
      type_(), session_buffer_(), server_(server) {
  CHECK_EQ(evutil_secure_rng_init(), 0);
  char buf[8];
  evutil_secure_rng_get_bytes(&buf, sizeof(buf));
  char hex_buf[16];
  for (uint64 i = 0; i < sizeof(buf); ++i) {
    uint8 lower_bits = (buf[i] & 0xF0) >> 4;
    uint8 upper_bits = buf[i] & 0x0F;

    hex_buf[i * 2] = ToHexChar(lower_bits);
    hex_buf[i * 2 + 1] = ToHexChar(upper_bits);
  }
  session_buffer_.set_sid(std::string(hex_buf, sizeof(hex_buf)));
  session_buffer_.set_cur_back_channel_flushed_index(-1);
  evutil_gettimeofday(&last_back_channel_write_activity_, NULL);
}

void Session::Write(const std::vector<std::string> &msg) {
  std::lock_guard<sg::base::thread::Mutex> lock(mutex_);
  WriteInternal(msg);
}

void Session::SetBackChannelRequest(es::Request *request) {
  CHECK_NOTNULL(request);
  std::lock_guard<std::recursive_mutex> conn_lock(
      *request->connection()->mutex());
  std::lock_guard<sg::base::thread::Mutex> lock(mutex_);
  if (request_ != NULL) {
    LOG(ERROR) << "duplicate back channel. discarding request_: " << request_;
    request_->EndChunk("");
  }
  evutil_gettimeofday(&back_channel_start_, NULL);
  request_ = request;

  request->GetQueryParam("TYPE", &type_);

  if (type_ == "html") {
    session_buffer_.set_padding(PaddingType::SCRIPT);
  } else {
    session_buffer_.set_padding(PaddingType::LENGTH);
  }
  RequestHelper::PaddingType padding =
      session_buffer_.padding() == PaddingType::SCRIPT ?
      RequestHelper::SCRIPT : RequestHelper::LENGTH;

  RequestHelper request_helper(request_);
  CHECK(request_helper.StartChunk(padding));

  int64 aid;
  if (request->GetQueryParamAsInt64("AID", &aid)) {
    DropBufferToArrayId(aid);
  }

  std::string ci;
  request_->GetQueryParam("CI", &ci);
  const bool buffered_proxy = ci == "1";

  if (session_buffer_.buffer_size() == 0 && !buffered_proxy) {
    // no pending messages, send a noop to initiate the new back channel
    JsonArray session_info;
    session_info.AppendString("noop");
    std::vector<std::string> write_vec;
    write_vec.push_back(session_info.Serialize());
    WriteInternal(write_vec);
  } else {
    // flush pending messages to initiate the new back channel
    FlushBufferInternal(NULL);
  }

  if (padding == RequestHelper::SCRIPT && !buffered_proxy) {
    // to match the gmail server, send kIePadding after the current round of
    // messages has been flushed. this potentially results in the messages
    // being processed sooner in IE < 10.
    CHECK(request_helper.SendChunk(kIePadding, RequestHelper::NONE));
  }
}

void Session::OnBackChannelDisconnect() {
  std::lock_guard<sg::base::thread::Mutex> lock(mutex_);
  if (request_ && !request_->IsFinished()) {
    request_->EndChunk("");
  }
  request_ = NULL;
}

std::string Session::GetSid() {
  std::lock_guard<sg::base::thread::Mutex> lock(mutex_);
  return session_buffer_.sid();
}

void Session::Stop() {
  std::lock_guard<sg::base::thread::Mutex> lock(mutex_);
  if (request_ != NULL) {
    RequestHelper request_helper(request_);
    RequestHelper::PaddingType padding =
        session_buffer_.padding() == PaddingType::SCRIPT ?
        RequestHelper::SCRIPT : RequestHelper::LENGTH;

    // TODO(ahochhaus)
    CHECK(request_helper.Send(es::OK, "[\"stop\"]", padding));
  }
  timer_event_.Stop();
  // TODO(ahochhaus)
  // SessionManager::Remove(this);
}

void Session::FlushBuffer(es::Request *request) {
  std::lock_guard<sg::base::thread::Mutex> lock(mutex_);
  FlushBufferInternal(request);
}

uint64 Session::last_array_id() {
  std::lock_guard<sg::base::thread::Mutex> lock(mutex_);
  return session_buffer_.buffer_start_array_id() +
      session_buffer_.buffer_size() - 1;
}

bool Session::back_channel_present() {
  std::lock_guard<sg::base::thread::Mutex> lock(mutex_);
  return request_ != NULL;
}

uint64 Session::outstanding_back_channel_bytes() {
  std::lock_guard<sg::base::thread::Mutex> lock(mutex_);
  return session_buffer_.outstanding_back_channel_bytes();
}

char Session::ToHexChar(uint8 value) {
  CHECK_LT(value, 16);
  if (value < 10) {
    return static_cast<char>(value + '0');
  }
  return static_cast<char>(value - 10 + 'A');
}

void Session::WriteInternal(const std::vector<std::string> &msg) {
  CHECK(mutex_.is_locked_in_current_thread());
  for (std::vector<std::string>::const_iterator i = msg.begin();
       i < msg.end(); ++i) {
    session_buffer_.add_buffer(*i);
    session_buffer_.set_outstanding_back_channel_bytes(
        session_buffer_.outstanding_back_channel_bytes() + i->length());
  }
  FlushBufferInternal(NULL);
}

void Session::FlushBufferInternal(es::Request *request) {
  CHECK(mutex_.is_locked_in_current_thread());
  // When a new session is being created it is necessary to flush the session
  // buffer down the forward channel.
  es::Request *output_request = request ? request : request_;

  int cur_back_channel_flushed_index =
      session_buffer_.cur_back_channel_flushed_index();

  if (output_request != NULL &&
      session_buffer_.buffer_size() > cur_back_channel_flushed_index + 1) {
    RequestHelper request_helper(output_request);

    JsonArray resp_list;
    for (int i = cur_back_channel_flushed_index + 1;
         i < session_buffer_.buffer_size(); ++i) {
      JsonArray *resp = resp_list.AppendArray();
      resp->AppendNumber(session_buffer_.buffer_start_array_id() +
                         cur_back_channel_flushed_index + 1);
      resp->AppendBytes(session_buffer_.buffer(i));
      ++cur_back_channel_flushed_index;
    }
    session_buffer_.set_cur_back_channel_flushed_index(
        cur_back_channel_flushed_index);

    if (request) {
      CHECK(request_helper.Send(es::OK,
                                resp_list.Serialize(),
                                RequestHelper::LENGTH));
    } else {
      RequestHelper::PaddingType padding =
          session_buffer_.padding() == PaddingType::SCRIPT ?
          RequestHelper::SCRIPT : RequestHelper::LENGTH;

      CHECK(request_helper.SendChunk(resp_list.Serialize(), padding));

      struct timeval now;
      evutil_gettimeofday(&now, NULL);
      const bool backchannel_timeout =
          now.tv_sec - back_channel_start_.tv_sec > kBackChannelThreshold;
      std::string ci;
      output_request->GetQueryParam("CI", &ci);
      const bool buffered_proxy = ci == "1";
      if (backchannel_timeout || buffered_proxy) {
        LOG(INFO) << "closing backchannel request: " << request_;
        CHECK(request_helper.EndChunk(padding));
        request_ = NULL;
      }
    }
    evutil_gettimeofday(&last_back_channel_write_activity_, NULL);
    SetupNoopCallback();
  }
}

void Session::DropBufferToArrayId(const uint64 array_id) {
  CHECK(mutex_.is_locked_in_current_thread());
  session_buffer_.set_cur_back_channel_flushed_index(-1);
  // nothing to drop
  if (session_buffer_.buffer_start_array_id() > array_id) {
    return;
  }
  uint64 drop_length = 0;
  uint64 drop_to_index = array_id - session_buffer_.buffer_start_array_id();
  for (uint64 i = 0; i <= drop_to_index; ++i) {
    drop_length += session_buffer_.buffer(i).length();
  }
  google::protobuf::RepeatedPtrField<std::string> new_buffer;
  for (int i = drop_to_index + 1; i < session_buffer_.buffer_size(); ++i) {
    auto s = new_buffer.Add();
    *s = session_buffer_.buffer(i);
  }
  session_buffer_.mutable_buffer()->CopyFrom(new_buffer);
  session_buffer_.set_outstanding_back_channel_bytes(
      session_buffer_.outstanding_back_channel_bytes() - drop_length);
  session_buffer_.set_buffer_start_array_id(array_id + 1);
}

void Session::SetupNoopCallback() {
  timer_event_.Reset(kNoopInterval);
}

void Session::NoopCallback() {
  struct timeval now;
  evutil_gettimeofday(&now, NULL);
  if (now.tv_sec - last_back_channel_write_activity_.tv_sec >
      kSessionInactivityThreshold) {
    timer_event_.Stop();
    std::string sid = GetSid();
    server_->CloseSession(sid);
    SessionManager::Remove(sid);
    return;
  } else if (request_ != NULL) {
    JsonArray session_info;
    session_info.AppendString("noop");
    std::vector<std::string> write_vec;
    write_vec.push_back(session_info.Serialize());
    Write(write_vec);
  }
  SetupNoopCallback();
}

}  // namespace comet
}  // namespace server
}  // namespace http
}  // namespace net
}  // namespace sg
